package com.lagou.hbase.client;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

/**
 * @Author: Hugh
 * @Date: 2020/9/17
 */
public class HbaseClientHomeWork {

    Configuration conf = null;
    Connection conn = null;

    @Before
    public void init() throws IOException {
        //获取一个配置文件对象
        conf = HBaseConfiguration.create();
        //由于启动的是mac的zk，需要配置本地的zk
        conf.set("hbase.zookeeper.quorum", "192.168.1.3");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        //通过conf获取到hbase集群的连接
        conn = ConnectionFactory.createConnection(conf);
    }

    //创建一张hbase表
    @Test
    public void createTable() throws IOException {
        //获取HbaseAdmin对象用来创建表
        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
        //创建Htabledesc描述器，表描述器
        final HTableDescriptor userRela = new HTableDescriptor(TableName.valueOf("user_relationship"));
        //指定列族
        userRela.addFamily(new HColumnDescriptor("friends"));
        admin.createTable(userRela);
        System.out.println("worker表创建成功！！");
    }

    /**
     * 初始化数据
     * @throws IOException
     */
    @Test
    public void putData() throws IOException {
        //需要获取一个table对象
        final Table userRela = conn.getTable(TableName.valueOf("user_relationship"));

        //准备list<puts>,可以执行批量插入
        List<Put> list = Arrays.asList(
                new Put(Bytes.toBytes("uuid1")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid2"), Bytes.toBytes("uuid2")),
                new Put(Bytes.toBytes("uuid1")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid3"), Bytes.toBytes("uuid3")),
                new Put(Bytes.toBytes("uuid1")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid4"), Bytes.toBytes("uuid4")),
                new Put(Bytes.toBytes("uuid1")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid5"), Bytes.toBytes("uuid5")),
                new Put(Bytes.toBytes("uuid1")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid6"), Bytes.toBytes("uuid6")),

                new Put(Bytes.toBytes("uuid2")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid1"), Bytes.toBytes("uuid1")),
                new Put(Bytes.toBytes("uuid2")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid3"), Bytes.toBytes("uuid3")),
                new Put(Bytes.toBytes("uuid2")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid4"), Bytes.toBytes("uuid4")),
                new Put(Bytes.toBytes("uuid2")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid5"), Bytes.toBytes("uuid5")),
                new Put(Bytes.toBytes("uuid2")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid6"), Bytes.toBytes("uuid6")),

                new Put(Bytes.toBytes("uuid3")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid1"), Bytes.toBytes("uuid1")),
                new Put(Bytes.toBytes("uuid3")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid2"), Bytes.toBytes("uuid2")),
                new Put(Bytes.toBytes("uuid3")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid4"), Bytes.toBytes("uuid4")),
                new Put(Bytes.toBytes("uuid3")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid5"), Bytes.toBytes("uuid5")),
                new Put(Bytes.toBytes("uuid3")).addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid6"), Bytes.toBytes("uuid6"))
        );
        userRela.put(list);
        //关闭table对象
        userRela.close();
        System.out.println("插入数据到user_relationship表成功！！");
    }

    //查询数据
    @Test
    public void getData() throws IOException {
        //准备table对象
        final Table userRela = conn.getTable(TableName.valueOf("user_relationship"));
        print("uuid1", userRela);
        print("uuid2", userRela);
        print("uuid3", userRela);
        isPresent("uuid2", userRela, "uuid3");
        userRela.close();
    }

    //删除一条数据
    @Test
    public void deleteData() throws IOException {
        //hbase(main):022:0> delete 'user_relationship', 'uuid1', 'friends:uuid3'
        //hbase(main):022:0> delete 'user_relationship', 'uuid3', 'friends:uuid1'


        //需要获取一个table对象
        final Table userRela = conn.getTable(TableName.valueOf("user_relationship"));

        final Delete delete1 = new Delete(Bytes.toBytes("uuid1"));
        delete1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid2"));
        userRela.delete(delete1);
        List<Cell> cells = delete1.getFamilyCellMap().get(
                Bytes.toBytes("friends"));
        byte[] qualifierArray = cells.get(0).getQualifierArray();
        String needDeletRowKey = new String(qualifierArray);
        userRela.close();
        System.out.println("删除数据成功！！");
    }

    private void print(String rowkey, Table userRela)throws IOException {
        //准备get对象
        final Get get = new Get(Bytes.toBytes(rowkey));
        //指定查询某个列族或者列
        get.addFamily(Bytes.toBytes("friends"));
        //执行查询
        final Result result = userRela.get(get);
        //获取到result中所有cell对象
        final Cell[] cells = result.rawCells();
        System.out.println("rowkey-->" + rowkey + " friends:");
        //遍历打印
        for (Cell cell : cells) {
            final String f = Bytes.toString(CellUtil.cloneFamily(cell));
            final String column = Bytes.toString(CellUtil.cloneQualifier(cell));
            final String value = Bytes.toString(CellUtil.cloneValue(cell));
            System.out.println(value);
        }
    }

    private Boolean isPresent(String rowkey, Table userRela, String info)throws IOException {
        //准备get对象
        final Get get = new Get(Bytes.toBytes(rowkey));
        //指定查询某个列族或者列
        get.addFamily(Bytes.toBytes("friends"));
        get.addColumn(Bytes.toBytes("friends"), Bytes.toBytes(info));
        //执行查询
        final Result result = userRela.get(get);
        //获取到result中所有cell对象
        final Cell[] cells = result.rawCells();
        final String value = Bytes.toString(CellUtil.cloneValue(cells[0]));
        System.out.println(value);
        return null != cells && cells.length > 0;
    }

    //释放连接
    @After
    public void realse() throws IOException {
        //需要获取一个table对象
        final Table userRela = conn.getTable(TableName.valueOf("user_relationship"));

        userRela.close();
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
